package org.simple.webserver.server;

import lombok.Setter;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscription;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.server.reactive.AbstractServerHttpResponse;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Mono;

import java.io.BufferedWriter;
import java.io.OutputStreamWriter;
import java.net.Socket;
import java.nio.charset.StandardCharsets;

/**
 * @author HaiLang
 * @date 2021/8/22 10:21 下午
 */
@Setter
public class MyServerHttpResponse extends AbstractServerHttpResponse {

    private Socket socket;

    public MyServerHttpResponse() {
        this(new DefaultDataBufferFactory());
    }

    public MyServerHttpResponse(DataBufferFactory dataBufferFactory) {
        super(dataBufferFactory);
    }

    public MyServerHttpResponse(DataBufferFactory dataBufferFactory, HttpHeaders headers) {
        super(dataBufferFactory, headers);
    }

    public MyServerHttpResponse(Socket socket) {
        this();
        this.socket = socket;
    }

    @Override
    public <T> T getNativeResponse() {
        return (T) this;
    }

    @Override
    protected Mono<Void> writeWithInternal(Publisher<? extends DataBuffer> body) {
        System.out.println("writeWithInternal");
        return writeAndFlushWithInternal(Mono.just(body));
    }

    @Override
    protected Mono<Void> writeAndFlushWithInternal(Publisher<? extends Publisher<? extends DataBuffer>> body) {
        System.out.println("writeAndFlushWithInternal");
        Mono.fromDirect(body).subscribe(new BaseSubscriber<Publisher<? extends DataBuffer>>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                subscription.request(1);
            }

            @Override
            protected void hookOnNext(Publisher<? extends DataBuffer> value) {
                System.out.println("hookOnNext");
                value.subscribe(new BaseSubscriber<DataBuffer>() {
                    @Override
                    protected void hookOnNext(DataBuffer value) {
                        try {
                            BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
                            bw.write("HTTP/1.1 200 OK\n");
                            bw.write("Content-Type: text/plain; charset=UTF-8\n");
                            bw.write("\n");
                            //真正的内容
                            bw.write(value.toString(StandardCharsets.UTF_8));
                            //必须在请求读取和响应写入都处理完毕之后才可以调用close方法，将输入流关闭也会导致输出流不可用
                            bw.close();
                            socket.close();
                        } catch (Exception e) {
                        }
                    }
                });
            }
        });
        return Mono.empty();
    }

    @Override
    protected void applyStatusCode() {

    }

    @Override
    protected void applyHeaders() {

    }

    @Override
    protected void applyCookies() {

    }
}
